package com.pushtime.service.impl.search;

import com.alibaba.fastjson.JSONObject;
import com.pushtime.domain.HdfsLog;
import com.pushtime.repository.HdfsLogRepository;
import com.pushtime.service.search.HdfsDownloadService;
import com.pushtime.web.rest.vm.HdfsVM;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.time.Instant;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

@Service
@Transactional
public class HdfsDownloadServiceImpl implements HdfsDownloadService {

    private final RestTemplate restTemplate;

    @Value("${application.webhdfs.blockSize}")
    private Integer LENGTH;
    @Value("${application.webhdfs.url}")
    private String url;
    @Value("${application.webhdfs.path}")
    private String path;
    @Value("${application.webhdfs.autoMkdir}")
    private Boolean autoMkdir;

    private final HdfsLogRepository hdfsLogRepository;

    public HdfsDownloadServiceImpl(RestTemplate restTemplate, HdfsLogRepository hdfsLogRepository) {
        this.restTemplate = restTemplate;
        this.hdfsLogRepository = hdfsLogRepository;
    }


    /**
     * 创建下载
     *
     * @param hdfsVM
     * @return
     */
    @Override
    public HdfsLog createHdfsDownload(HdfsVM hdfsVM) {
        // 设置下载初始信息

        String dstFile = hdfsVM.getDstFile();
        // 创建文件夹
        if (autoMkdir) {
            String[] dirs = dstFile.split("/");
            String dirInfo = StringUtils.join(Arrays.copyOf(dirs, dirs.length - 1), '/');
            File file = new File(dirInfo);
            file.mkdirs();
        }

        HdfsLog hdfsLog = new HdfsLog();
        hdfsLog.setSrc(hdfsVM.getSrcFile());
        hdfsLog.setDst(dstFile);
        hdfsLog.createdDate(Instant.now());
        hdfsLog.setProcess(0D);
        hdfsLog.setCurrentSize(0L);
        hdfsLog.setName("download");

        Map<String, Object> params = new HashMap<>();
        params.put("op", "GETFILESTATUS");
        System.out.println(url + path);
        System.out.println(hdfsVM.getSrcFile());
        try {
            ResponseEntity<String> responseEntity = restTemplate.getForEntity(url + path
                + hdfsVM.getSrcFile() + "?op={op}", String.class, params);
            JSONObject jsonObject = JSONObject.parseObject(responseEntity.getBody());
            JSONObject fileStatus = jsonObject.getJSONObject("FileStatus");
            if (fileStatus != null) {
                hdfsLog.setState(1);
                hdfsLog.setTotalSize(fileStatus.getLong("length"));
            } else {
                //  异常，通知失败
                hdfsLog.setState(2);
                hdfsLog.setTotalSize(0L);
                hdfsLog.setDesc("server response error!");
            }
        } catch (Exception e) {
            hdfsLog.setState(2);
            hdfsLog.setTotalSize(0L);
            hdfsLog.setDesc(e.getMessage());
        }
        return hdfsLogRepository.save(hdfsLog);
    }

    /**
     * 创建 HDFS 下载
     *
     * @param hdfsLog
     */
    @Override
    public ResponseEntity<byte[]> startDownload(HdfsLog hdfsLog, long offset, int length) {
        // 选择保存文件地址
        Map<String, Object> params = new HashMap<>();
        params.put("op", "OPEN");
        params.put("offset", offset);
        params.put("length", length);
        return restTemplate.getForEntity(url + path
            + hdfsLog.getSrc() + "?op={op}&offset={offset}&length={length}", byte[].class, params);
    }

    @Override
    public void execute(HdfsLog tmpLog, FileOutputStream fileOutputStream) throws IOException {
        ResponseEntity<byte[]> responseEntity = startDownload(tmpLog, tmpLog.getCurrentSize(), LENGTH);
        byte[] body = responseEntity.getBody();
        if (body != null) {
            fileOutputStream.write(body);
            tmpLog.setCurrentSize(tmpLog.getCurrentSize() + body.length);
            tmpLog.setProcess(100.0 * tmpLog.getCurrentSize() / tmpLog.getTotalSize());
            tmpLog = hdfsLogRepository.save(tmpLog);
        } else {
            tmpLog.setState(3);
            tmpLog = hdfsLogRepository.save(tmpLog);
        }
    }



    /**
     * 获取下载状态
     *
     * @param id 任务编号
     * @return 返回下载状态信息
     */
    @Override
    public Optional<HdfsLog> getStatus(Long id) {
        return hdfsLogRepository.findById(id);
    }

    /**
     * 创建同步下载接口
     * @param hdfsVM
     * @return
     */
    @Override
    public HdfsLog createSynchronizedHdfsDownload(HdfsVM hdfsVM) {
        HdfsLog hdfsLog = createHdfsDownload(hdfsVM);
        if (hdfsLog.getState() == 1) {
            try {
                FileOutputStream fileOutputStream = new FileOutputStream(new File(hdfsLog.getDst()));
                while (hdfsLog.getState().equals(1) &&
                    !(hdfsLog.getTotalSize()<= hdfsLog.getCurrentSize())) {
                    execute(hdfsLog, fileOutputStream);
                }
                fileOutputStream.close();
            } catch (IOException e) {
                hdfsLog.setDesc(e.getMessage());
                hdfsLogRepository.save(hdfsLog);
            }
        }
        return hdfsLog;
    }
}
